package com.pai4j.netty.server.service;
import com.pai4j.common.util.JsonUtil;
import com.pai4j.domain.vo.dto.im.UgcServerMessageDTO;
import com.pai4j.netty.server.enums.TopicNameEnum;
import com.pai4j.netty.server.rocketmq.producer.RocketMQProducer;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;

@Slf4j
public class IMMessageHandler {

    private static ChannelRegistryService channelRegistryService = new ChannelRegistryService();

    public boolean handleMessage(MessageExt msg) {


        String ugcMessage = new String(msg.getBody(), StandardCharsets.UTF_8);

        UgcServerMessageDTO ugcServerMessage = JsonUtil.fromJson(ugcMessage, UgcServerMessageDTO.class);

        // 消息接收人
        // 消息代理人模式（如果来自业务的消息有代理人，优先转发给代理人）
        String proxyReceiverId = ugcServerMessage.getProxyReceiverId();
        String realReceiverId = StringUtils.isNotBlank(proxyReceiverId) ? proxyReceiverId :
                ugcServerMessage.getReceiverId();
        // 获取消息接收人长链接所在节点信息
        String instance = channelRegistryService.getInstance(realReceiverId);
        try {
            SendResult sendResult = RocketMQProducer.send(TopicNameEnum.TOPIC_NODE_IM_SEND_MSG.getTopic(), instance, ugcMessage);
            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                // 直链消息推送成功
                log.info("直链消息推送成功 ===> tag:{}", instance);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


        return false;
    }


    /**
     * 真正的消息推送
     *
     * @param msg
     */
    public void doWriteMessage2Receiver(MessageExt msg) {

        String message = new String(msg.getBody(), StandardCharsets.UTF_8);
        log.info("consume message from netty node ===> {} ", message);
        UgcServerMessageDTO ugcServerMessageDTO =  JsonUtil.fromJson(message, UgcServerMessageDTO.class);
        String senderId = ugcServerMessageDTO.getSenderId();

        // 消息代理人模式（如果来自业务的消息有代理人，优先转发给代理人）
        String proxyReceiverId = ugcServerMessageDTO.getProxyReceiverId();
        String writeReceiverId = StringUtils.isNotBlank(proxyReceiverId) ? proxyReceiverId :
                                 ugcServerMessageDTO.getReceiverId();
        UserChannelManager.getChannel(writeReceiverId).writeAndFlush(new TextWebSocketFrame(message)).addListener((future) -> {
            if (future.isSuccess()) {
                log.info("push message to client ===> senderId:{}, receiverId: {} ", senderId, writeReceiverId);
            } else {
                // TODO 兜底策略
            }
        });
    }

}
